package com.bw.ad1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.OrderStatus;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

public class Test10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9", "test4" + System.currentTimeMillis()));

        //筛选订单数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String tableName = jsonObject.getString("table");
                if ("order_info".equals(tableName)) {
                    return true;
                }
                return false;
            }
        });
        // 变成JSON数据
        SingleOutputStreamOperator<JSONObject> mapStream = orderInfoDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> watermarkStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                // 时间戳16位
                return element.getJSONObject("data").getLong("create_time") / 1000;
            }
        }));


        /*

                订单ID   订单状态
                  1        50010
                  2        50010
                  3        50020
                  1        50020
         */
        
        watermarkStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(30))).process(new ProcessAllWindowFunction<JSONObject, OrderStatus, TimeWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<JSONObject, OrderStatus, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<OrderStatus> collector) throws Exception {
                Set<String> ids = new HashSet<>();
                int unpay = 0;
                int unSort = 0;
                int unWrite = 0;
                int unTake = 0;
                Iterator<JSONObject> iterator = iterable.iterator();
                while (iterator.hasNext()){
                    JSONObject orderInfo = iterator.next();
                    // 取出订单的ID
                    String id = orderInfo.getJSONObject("data").getString("id");
                    ids.add(id);
                    String status = orderInfo.getJSONObject("data").getString("status");

                    if ("60010".equals(status)){
                        unpay++;
                    }else if ("60040".equals(status)){
                        unSort++;
                    }else if ("60070".equals(status)){
                        unWrite++;
                    }else if ("60020".equals(status)){
                        unTake++;
                    }
                }
                // 一小时的订单数
                Double OrderCount = Double.valueOf(ids.size());

                System.out.println(unpay/OrderCount);
                System.out.println(unSort/OrderCount);
                System.out.println(unWrite/OrderCount);
                System.out.println(unTake/OrderCount);
                System.out.println(context.window().getStart());
            }
        });
        env.execute();
    }
}
